package com.cbh.amadeus.client.receiver.tool.impl;

import com.cbh.amadeus.client.domain.MessageRepositoryModel;
import com.cbh.amadeus.client.receiver.tool.MessageRepositoryModelTool;
import com.cbh.amadeus.common.utils.StringPool;
import lombok.extern.slf4j.Slf4j;

import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;

/**
 * @author ChenBiHui
 * @date 2025/5/8
 */
@Slf4j
public class DefaultMessageModelRepositoryTool implements MessageRepositoryModelTool {

    private final Set<MessageRepositoryModel> record = ConcurrentHashMap.newKeySet(64);
    private final Map<String, ReentrantLock> lock = new ConcurrentHashMap<>(64);
    private final AtomicLong sequence = new AtomicLong(0);


    @Override
    public String sequence() {
        return String.valueOf(sequence.incrementAndGet());
    }

    @Override
    public void commit(MessageRepositoryModel model) {
        record.add(model);
    }

    @Override
    public void adjust(String tenant, String topic, String messageId, MessageRepositoryModel.MessageRepositoryEnum status) {
        MessageRepositoryModel target = record.stream()
                .filter(item -> item.getTenant().equals(tenant))
                .filter(item -> item.getTopic().equals(topic))
                .filter(item -> item.getMessageId().equals(messageId))
                .findFirst()
                .orElse(null);
        if (target == null) {
            return;
        }
        try {
            if (target.getStatus().equals(MessageRepositoryModel.MessageRepositoryEnum.ACTIVE.name())) {
                lock.computeIfAbsent(key(topic, messageId), key -> new ReentrantLock()).lock();
                if (target.getStatus().equals(MessageRepositoryModel.MessageRepositoryEnum.ACTIVE.name())) {
                    target.setStatus(status.name());
                    target.setUpdateTime(LocalDateTime.now());
                }
            }
        } finally {
            Optional.ofNullable(lock.remove(key(topic, messageId)))
                    .ifPresent(ReentrantLock::unlock);
        }
    }

    @Override
    public void reject(String tenant, String topic, String messageId) {
        try {
            lock.computeIfAbsent(key(topic, messageId), key -> new ReentrantLock()).lock();
            record.stream()
                    .filter(item -> item.getTenant().equals(tenant))
                    .filter(item -> item.getTopic().equals(topic))
                    .filter(item -> item.getMessageId().equals(messageId))
                    .forEach(item -> item.setReject(item.getReject() + 1));
        } finally {
            Optional.ofNullable(lock.remove(key(topic, messageId)))
                    .ifPresent(ReentrantLock::unlock);
        }
    }

    @Override
    public List<MessageRepositoryModel> list(Long limit, String tenant, MessageRepositoryModel.MessageRepositoryEnum type) {
        return record.stream()
                .filter(item -> item.getTenant().equals(tenant))
                .filter(item -> item.getStatus().equals(type.name()))
                .limit(limit)
                .collect(Collectors.toList());
    }

    private String key(String topic, String messageId) {
        return topic + StringPool.COMMA + messageId;
    }
}
